#pragma once
#include "Consumer.hpp"
#include "muduo/net/TcpConnection.h"
#include "../third/protobuf/codec.h"
#include "../third/protobuf/dispatcher.h"
#include "../mqcommon/Log.hpp"
#include "../mqcommon/Helper.hpp"
#include "../mqcommon/msg.pb.h"
#include "../mqcommon/proto.pb.h"
#include <iostream>
#include <mutex>
#include <condition_variable>
#include <unordered_map>
using namespace std;
namespace mq
{
    using ProtobufCodecPtr = shared_ptr<ProtobufCodec>;
    using basicCommonResponsePtr = shared_ptr<mq::basicCommonResponse>;
    class Channel
    {
    public:
        using ChannelPtr = shared_ptr<Channel>;
        Channel(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec)
            : _conn(conn), _codec(codec), _cid(mq::UuidHelper::uuid())
        {
        }
        ~Channel()
        {
            basicCancel();
        }
        string cid()
        {
            return _cid;
        }
        bool openChannel()
        {
            mq::openChannelRequest req;
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn, req);
            basicCommonResponsePtr crp = waitResponse(rid);
            return crp->ok();
        }
        void closeChannel()
        {
            mq::closeChannelRequest req;
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn, req);
            waitResponse(rid);
        }
        bool declareExchange(const string &name, ExchangeType type, bool durable, bool auto_delete,
                             google::protobuf::Map<string, string> &args)
        {
            mq::declareExchangeRequest req;
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_ename(name);
            req.set_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            _codec->send(_conn, req);
            basicCommonResponsePtr crp = waitResponse(rid);
            return crp->ok();
        }
        void deleteExchange(const string &name)
        {
            mq::deleteExchangeRequest req;
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_ename(name);
            _codec->send(_conn, req);
            waitResponse(rid);
        }
        bool declareQueue(const string &qname, bool qdurable, bool qexclusive, bool qauto_delete,
                          google::protobuf::Map<string, string> &qargs)
        {
            mq::declareQueueRequest req;
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_qname(qname);
            req.set_exclusive(qexclusive);
            req.set_durable(qdurable);
            req.set_auto_delete(qauto_delete);
            req.mutable_args()->swap(qargs);
            _codec->send(_conn, req);
            basicCommonResponsePtr crp = waitResponse(rid);
            return crp->ok();
        }
        void deleteQueue(const string &qname)
        {
            mq::deleteQueueRequest req;
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_qname(qname);
            _codec->send(_conn, req);
            waitResponse(rid);
        }
        bool queueBind(const string &ename, const string &qname, const string &key)
        {
            mq::BindRequest req;
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_ename(ename);
            req.set_qname(qname);
            req.set_binding_key(key);
            _codec->send(_conn, req);
            basicCommonResponsePtr crp = waitResponse(rid);
            return crp->ok();
        }
        void queueUnBind(const string &ename, const string &qname)
        {
            mq::unBindRequest req;
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_ename(ename);
            req.set_qname(qname);
            _codec->send(_conn, req);
            waitResponse(rid);
        }
        void basicPublish(const string &ename, const BasicProperties *bp, const string &body)
        {
            basicPublishRequest req;
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_ename(ename);
            req.set_body(body);
            if (bp != nullptr)
            {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_routing_key(bp->routing_key());
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
            }
            _codec->send(_conn, req);
            mq::basicCommonResponsePtr resp = waitResponse(rid);
            // DLOG("%s", resp->ok() == true ? "true" : "false");
        }
        void basicAck(const string &msgid)
        {
            if (_consumer.get() == nullptr)
            {
                DLOG("未订阅任何消息");
                return;
            }
            mq::basicAckRequest req;
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_qname(_consumer->_qname);
            req.set_msg_id(msgid);
            _codec->send(_conn, req);
            waitResponse(rid);
        }
        void basicCancel()
        {
            if (_consumer.get() == nullptr)
            {
                DLOG("未订阅任何消息");
                return;
            }
            mq::basicCancelRequest req;
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_tag(_consumer->_tag);
            req.set_qname(_consumer->_qname);
            _codec->send(_conn, req);
            waitResponse(rid);
        }
        bool basicConsume(const string &tag, const string &qname, bool auto_ack, const ConsumerCallback &cb)
        {
            if (_consumer.get() != nullptr)
            {
                DLOG("当前信道已订阅其他队列消息！");
                return false;
            }
            mq::basicConsumeRequest req;
            string rid = UuidHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_tag(tag);
            req.set_qname(qname);
            req.set_auto_ack(auto_ack);
            _codec->send(_conn, req);
            // DLOG("已发送建立信道请求:%s", _cid.c_str());
            basicCommonResponsePtr crp = waitResponse(rid);
            if (crp->ok() == false)
            {
                DLOG("添加订阅失败");
                return false;
            }
            _consumer = make_shared<Consumer>(tag, qname, auto_ack, cb);
            return crp->ok();
        }

        void putBasicResponse(const basicCommonResponsePtr &resp)
        {
            unique_lock<mutex> lock(_mutex);
            _base_resp.insert(make_pair(resp->rid(), resp));
            _cond.notify_all();
        }
        void consume(const basicConsumeResponsePtr &resp)
        {
            if (_consumer.get() == nullptr)
            {
                DLOG("消息处理时，未找到订阅者信息！");
                return;
            }
            if (_consumer->_tag != resp->tag())
            {
                DLOG("消息处理时，消费者标识不一致！");
                return;
            }
            _consumer->_callback(resp->tag(), resp->mutable_properties(), resp->body());
        }

    private:
        basicCommonResponsePtr waitResponse(const string &rid)
        {
            unique_lock<mutex> lock(_mutex);
            _cond.wait(lock, [&rid, this]()
                       { return _base_resp.find(rid) != _base_resp.end(); });
            basicCommonResponsePtr crp = _base_resp[rid];
            _base_resp.erase(rid);
            return crp;
        }

    private:
        string _cid;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        Consumer::ConsumerPtr _consumer;
        mutex _mutex;
        condition_variable _cond;
        unordered_map<string, basicCommonResponsePtr> _base_resp;
    };
    class ChannelManager
    {
    public:
        using ChannelManagerPtr = shared_ptr<ChannelManager>;
        ChannelManager() {}
        Channel::ChannelPtr create(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec)
        {
            unique_lock<mutex> lock(_mutex);
            Channel::ChannelPtr cp = make_shared<Channel>(conn, codec);
            _consumers.insert(make_pair(cp->cid(), cp));
            return cp;
        }
        void remove(const std::string &cid)
        {
            unique_lock<mutex> lock(_mutex);
            _consumers.erase(cid);
        }
        Channel::ChannelPtr get(const std::string &cid)
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _consumers.find(cid);
            if (it == _consumers.end())
            {
                DLOG("目标信道不存在: %s", cid.c_str());
                return Channel::ChannelPtr();
            }
            return it->second;
        }

    private:
        mutex _mutex;
        unordered_map<string, Channel::ChannelPtr> _consumers;
    };
}